package cn.edu.flink.scala.tutorial.window

import cn.edu.flink.scala.tutorial.source.ClickSource
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger

object TriggerTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    env.setParallelism(1)

    val eventsDS = env.addSource(new ClickSource())
      .assignAscendingTimestamps(_.timestamp)

    eventsDS
      .keyBy(_.user)
      .window(TumblingEventTimeWindows.of(Time.seconds(10)))
      .trigger(EventTimeTrigger.create())
  }
}
